/*
    实现异步工作器
        1. 异步线程使用时，会使用多线程来保证运行速度，所以锁和条件变量来保证线程安全
        2. 在多线程工作时，采用生产者和消费者的思想（使用两个线程和两个缓冲区），一个线程负责将数据放入输入缓冲区，另一个线程负责将输出缓冲区内的数据进行落地
            输入缓冲区有了数据之后交换线程，那么生产者线程继续向缓冲区进行写入数据，消费者线程落地数据
*/

#ifndef __M_LOOPER_H__
#define __M_LOOPER_H__

#include "buffer.hpp"
#include <mutex>
#include <thread>
#include <condition_variable>
#include <functional>
#include <atomic>

namespace Log
{
    // void是被调用函数的返回类型，()内的是被调用函数的形参
    using Functor = std::function<void(Buffer&)>;
    enum class AsyncType
    {
        ASYNC_SAFE,     // 安全状态，表示缓冲区满了则阻塞，避免资源耗尽的风险
        ASYNC_UNSAFE    // 不考虑资源耗尽的问题，无限扩容，常用于测试
    };
    class AsyncLooper
    {
    public:
        using prt = std::shared_ptr<AsyncLooper>;
        AsyncLooper(const Functor &cb,AsyncType looper_type = AsyncType::ASYNC_SAFE)
            :_stop(true)
            ,_looper_type(looper_type)
            ,_looper_callback(cb)
            ,_thread(std::thread(&AsyncLooper::threadEntry,this))   // 一定要指明所用的是类内函数，否则编译器会认为这是全局函数
        {}
        ~AsyncLooper()
        {
            stop();
        }

        void stop()
        {
            _stop = false;
            // 唤醒所有工作线程
            _con_cond.notify_all();

            // 等待工作线程的结束
            _thread.join();
        }

        void push(const char* data,size_t len)
        {
            // 1. 判断数据长度是否能够放入缓冲区，如果不能就阻塞
            std::unique_lock<std::mutex> lock(_mutex);
            if(_looper_type == AsyncType::ASYNC_SAFE)
                _pro_cond.wait(lock,[&](){ return len <= _pro_buf.writeAbleSize();});
    
            _pro_buf.push(data,len);

            // 2. 唤醒消费者处理数据
            _con_cond.notify_all();
        }
    private:
        // 线程入口函数
        void threadEntry()
        {
            while (1)
            {
                {
                    // 1. 判断生产者缓冲区是否有数据，如果有则交换，否则阻塞
                    std::unique_lock<std::mutex> lock(_mutex);
                    // 退出标志被设置且生产缓冲区无数据，这时退出，可能会造成生产缓冲区中有数据，但是没有完全被处理
                    if(!_stop && _pro_buf.empty())
                        break;
                    _con_cond.wait(lock,[&](){ return !_stop || !_pro_buf.empty(); });
                    _con_buf.swap(_pro_buf);

                    // 2. 唤醒生产者条件变量，继续生产数据
                    if(_looper_type == AsyncType::ASYNC_SAFE)
                        _pro_cond.notify_all();
                }
                
                // 3. 消费者缓冲区进行输出
                _looper_callback(_con_buf);

                // 4. 重置消费者缓冲区
                _con_buf.reset();
            }
        }
    
    private:
        AsyncType _looper_type;
        Functor _looper_callback;
        std::atomic<bool> _stop;                // 退出标志
        Buffer _pro_buf;                        // 生产者缓冲区
        Buffer _con_buf;                        // 消费者缓冲区
        std::mutex _mutex;
        std::condition_variable _pro_cond;      // 生产者条件变量
        std::condition_variable _con_cond;      // 消费者条件变量
        std::thread _thread;                    // 异步工作器对应的工作线程
    };
}

#endif